• Overview

• Terminology

• I/O Overhead

• Missed Aggregations

  • Server Outage

  • Failed Aggregation

  • Late Measurement Reports

• Schema Changes

  • Reducing I/O Overhead

  • Handling Missed Aggregations

• Upgrades

Overview

This document explains schema changes that aim to reduce I/O overhead during aggregation and to handle missed aggregations. The latest RHQ release as of this writing is 4.9.0. The latest JON release is 3.2.0.

Terminology

This section establishes some terminology to help make some of the concepts discussed more understandable and to avoid potential confusion.

Row

Unless otherwise stated a row will refer to a CQL row. Remember though that CQL transposes rows and columns, making them look more similar to relational rows and columns.

Partition

A partition refers to one physical row in a table or column family. A partition may be comprised of multiple CQL rows.

Primary Key

Uniquely identifies a row. It can consist of one or more columns.

Partition Key

The first column of the primary key. All rows having the same partition key are stored in the same partition and on the same node.

Clustering Columns

Any additional columns in the primary key after the partition key. Clustering columns define grouping and sorting within a partition.

Bucket

One of the metric data tables - raw_metrics, one_hour_metrics, six_hour_metrics, twenty_four_hour_metrics

I/O Overhead

We currently have the metrics_index table with the following schema,

CREATE TABLE metrics_index (
  bucket text,
  time timestamp,
  schedule_id int,
  PRIMARY KEY ((bucket, time), schedule_id)
)

We perform two writes when inserting raw data - one to the raw_metrics table and one to metrics_index. Suppose during the 10:00 hour we insert raw data for schedule id 100 having a timestamp of 10:02:00. The insert into metrics_index will look like,

INSERT INTO metrics_index (bucket, time, schedule_id)
VALUES (‘one_hour_metrics’, 10:00, 100)

The timestamp is rounded down to the start of the current time slice, 10:00. During aggregation, a single query is executed against metrics_index to obtain all of the schedule ids with raw data to be aggregated. Then for each schedule id, a query is executed against raw_metrics to fetch the data to be aggregated into 1 hour metrics.

If the query against metrics_index returns N measurement schedule ids, then N + 1 queries are executed to aggregate raw data. In the worst case scenario where we also have to aggregate 1 hour and 6 hour data, 3N + 3 queries are executed.

This does not scale well as the number of schedules increases as it leads to a lot of I/O overhead on the storage node(s).

Missed Aggregations

There are a few scenarios in which we could fail to aggregate metric data.

Schema Changes

Two tables are being added - metrics_cache and metrics_cache_index. The metrics_index table will be dropped. Here is the schema.

CREATE TABLE metrics_cache (
  bucket text,
  time_slice timestamp,
  start_schedule_id int,
  schedule_id int,
  time timestamp,
  value map<int, double>,
  PRIMARY KEY ((bucket, time_slice, start_schedule_id), schedule_id, time)
);

CREATE TABLE metrics_cache_index (
  bucket text,
  time_slice timestamp,
  partition int,
  start_schedule_id int,
  time timestamp,
  set<int> schedule_ids,
  PRIMARY KEY ((bucket, time_slice, partition), start_schedule_id, time)
);

When we insert metric data, we write to raw_metrics, metrics_cache, and metrics_cache_index. metrics_cache and metrics_cache_index are only queried during aggregation and during server startup.

We store data for multiple schedules per partition in metrics_cache. We will refer to the number of schedules per partition as the partition block size (PBS). The block size is configured at server startup.

We will look at an example of storing raw data to explain the new tables. Suppose we have the following raw data and PBS = 10.

{scheduleId: 100, timestamp: 10:18:00, value: 3.14},
{scheduleId: 221, timestamp: 10:16:00, value: 84},
{scheduleId: 366, timestamp: 10:09:00, value: 2.17}

Let’s first look at the INSERT statements for metrics_cache.

INSERT INTO metrics_cache
  (bucket, time_slice, start_schedule_id, schedule_id, time, value)
VALUES (‘raw_metrics’, 10:00, 100, 100, 10:18, {3: 3.14});

INSERT INTO metrics_cache
  (bucket, time_slice, start_schedule_id, schedule_id, time, value)
VALUES (‘raw_metrics’, 10:00, 220, 221, 10:16, {3: 84});

INSERT INTO metrics_cache
  (bucket, time_slice, start_schedule_id, schedule_id, time, value)
VALUES (‘raw_metrics’, 10:00, 360, 366, 10:09, {3: 2.17});

bucket identifies the historical table in which the data is stored. It will have as its value one of raw_metrics, one_hour_metrics, or six_hour_metrics depending on the type of metric data.

time_slice is derived by rounding down the timestamp to the start of the current time slice.

start_schedule_id will be determined by the values of scheduleId and PBS. It identifies a block of schedule IDs, e.g., 100 to 110, 220 to 230, 360 to 370.

time is the actual timestamp for the data being inserted.

value is stored as a map so that we can accommodate both raw and aggregate metrics.

Here are the INSERT statements for metrics_cache_index.

INSERT INTO metrics_cache_index
  (bucket, time_slice, partition, start_schedule_id, time)
VALUES (‘raw_metrics’, 10:00, 0, 100, 10:00);

INSERT INTO metrics_cache_index
  (bucket, time_slice, partition, start_schedule_id, time)
VALUES (‘raw_metrics’, 10:00, 0, 220, 10:00);

INSERT INTO metrics_cache_index
  (bucket, time_slice, partition, start_schedule_id, time)
VALUES (‘raw_metrics’, 10:00, 0, 360, 10:00);

bucket is one of the historical tables and has the same value as metrics_cache.bucket.

time_slice is the time slice in which the data is inserted. This will typically be the same as metrics_cache.time_slice but may differ in the event of late measurement reports.

partition is a number between 0 and n - 1 where n is the number of partitions that we want to split the bucket, time slice across. Initially and probably in many cases n will always be zero. If and when a single partition gets very big, it can lead to a hot spot on a node. If we find that that we have hit that point, we can increase the value of n. Clients will alternate the value of n on writes in a round robin fashion so that partitions grow at a similar rate. Partitions will be merged client side during aggregation.

start_schedule_id is the same as metrics_cache.start_schedule_id.

time is the time slice in which the data was collected. Under normal circumstances time and time_slice will have the same value. They will differ though in the cases of late measurement reports and failed aggregations.

INSERT INTO metrics_cache_index
  (bucket, time_slice, partition, start_schedule_id, time)
VALUES (‘raw_metrics’, 11:00, 0, 360, 10:00);

INSERT INTO metrics_cache_index
  (bucket, time_slice, partition, start_schedule_id, time, schedule_ids)
VALUES (‘raw_metrics’, 11:00, 0, 360, 09:00, {221});

INSERT INTO metrics_cache_index
  (bucket, time_slice, partition, start_schedule_id, time, schedule_ids)
VALUES (‘raw_metrics’, 11:00, 0, 360, 09:00, {366});

Upgrades

Schema changes are applied at installation (or upgrade) time. There could potentially be a large amount of data to migrate over to metrics_cache. This could make for a long-running, complicated upgrade process. A better solution is to only populate metrics_cache_index which should be very fast by comparison. We specify schedule_ids as we do for late measurement reports so that data is pulled from raw_metrics during the initial aggregation run. Subsequent aggregation runs will then use metrics_cache as previously described.